package cn.me.alphamq.broker.event.spi.listener.cluster;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.event.model.cluster.StartSyncEvent;
import cn.me.alphamq.broker.store.SlaveCtxManager;
import cn.me.alphamq.common.entity.resp.StartSyncResp;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.util.NettyUtil;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;

/**
 * broker 的主从复制中，开启同步事件的监听器
 *
 * @author f
 */
public class StartSyncListener implements Listener<StartSyncEvent> {

    @Override
    public void onReceive(StartSyncEvent event) throws Exception {
        String msgId = event.getMsgId();
        ChannelHandlerContext ctx = event.getCtx();

        String address = NettyUtil.getAddressFromCtx(ctx);
        SlaveCtxManager.put(address, ctx);
        NettyUtil.setCtxId(ctx, address);

        ctx.writeAndFlush(new TcpMsg(EventCodeEnum.START_SYNC_RESP, JSON.toJSONBytes(new StartSyncResp(msgId))));
    }
}
